package com.forchange.project.core.hotcategorytop10review.service

import com.forchange.project.core.TService
import com.forchange.project.core.bean.hotcategorytop10review.bean
import com.forchange.project.core.hotcategorytop10review.dao.HotCategoryTOP10DAOReview
import org.apache.spark.rdd.RDD


/**
 * @Author chencong
 * @Description 计算逻辑
 * @Date 4:08 下午 2021/6/27
 * @Param
 * */
class HotCategoryTOP10ServiceReview extends TService {
  private val hotCategoryTOP10DAOReview = new HotCategoryTOP10DAOReview

  override def analysis() = {
    //获取数据，数据已经被封装成一个一个的对象
    val UserRDD: RDD[bean.UserVisitAction] = hotCategoryTOP10DAOReview.getUserVisitction()
    val categoryToOneRDD: RDD[(String, (Int, Int, Int))] = UserRDD
      .flatMap(UserBean => {
        if (UserBean.click_category_id != "-1") {
          List((UserBean.click_category_id, (1, 0, 0)))
        } else if (UserBean.order_category_ids != "null") {
          val ids: Array[String] = UserBean.order_category_ids.split(",")
          ids.map(id => (id, (0, 1, 0)))
        } else if (UserBean.pay_category_ids != "null") {
          val ids: Array[String] = UserBean.pay_category_ids.split(",")
          ids.map(id => (id, (0, 0, 1)))
        }
        else {
          Nil
        }
      })
    val categorySumRDD: RDD[(String, (Int, Int, Int))] = categoryToOneRDD.reduceByKey {
      case ((click, order, pay), (clink1, order1, pay1)) => {
        (click + clink1, order + order1, pay + pay1)
      }
    }
    val result = categorySumRDD.sortBy(_._2, false) take (10)
    result
  }

  override def analysis(data: Any): Any = ???
}
